import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.kafka.common.metrics.Sensor;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Random;

public class SourceTest1_Collection {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<SensorReading> dataStream = env.addSource(new MySensor());


        dataStream.print();
        env.execute();
    }

    public static class MySensor implements SourceFunction<SensorReading>{

        private  boolean running = true;
        @Override
        public void run(SourceContext<SensorReading> sourceContext) throws Exception {
            Random random = new Random();
            HashMap<String,Double> sensorTempMap = new HashMap<String,Double>();
            for (int i = 0; i < 10; i++) {
                sensorTempMap.put("sensor_"+(i+1),60+random.nextGaussian()*20);
            }

            while(running){
                for (String sensorID : sensorTempMap.keySet()) {

                }
            }

        }

        @Override
        public void cancel() {

        }
    }
}


